package com.wzh.project.redis.consumer;

import java.util.List;

import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;

import com.wzh.project.common.Constants;
import com.wzh.project.redis.bean.Message;
import com.wzh.project.redis.bean.RedisMessage;
import com.wzh.project.redis.context.RedisContext;
import com.wzh.project.redis.handle.RedisReplyHandler;
import com.wzh.project.redis.utils.JedisUtils;
import com.wzh.project.service.DealService;
import com.wzh.project.utils.BinaryTransferUtils;

import redis.clients.jedis.Jedis;
import redis.clients.jedis.JedisCluster;
import redis.clients.jedis.JedisPool;

/**
 * 功能：从队列中获取消息
 * @author wzh
 *
 */
public class ConsumerThread implements Runnable {
	private static final Logger logger =LogManager.getLogger(ConsumerThread.class);
	//需要配置使用的redis与配置文件对应
	private String redisName;
	//监听队列的名字
	private String queueName;
	
	public void setQueueName(String queueName) {
		this.queueName = queueName;
	}

	public void setRedisName(String redisName) {
		this.redisName = redisName;
	}

	@Override
	public void run() {
		logger.debug("开始监听队列:"+queueName);
		while (true) {
    		RedisContext jc = RedisContext.getInstance();
    		Object o = null;
    		Jedis resource = null;
	        try {
	        	o = jc.getJedis(redisName);
	            if (o instanceof JedisPool) {
	            	JedisPool jedisPool=(JedisPool)o;
	            	resource = jedisPool.getResource();
	            	List<byte[]> list = resource.brpop(0, queueName.getBytes());
	            	dealMessage(list);
	            }else {
	            	JedisCluster jcr = (JedisCluster)o;
	            	List<byte[]> list = jcr.brpop(0, queueName.getBytes());
	            	dealMessage(list);
	            }
	        } catch (Exception e) {
	            logger.error("从频道获取消息错误，队列为："+queueName, e);
	        } finally {
	        	  if( o==null ) {
	        		  return;
	        	  }
	        	  if (o instanceof JedisPool) {
		            	if(resource!=null) {
		            		resource.close();
		            	}
		            }
	        }
		}
	}
	
	/**
	 * 消息处理
	 * @param messageList
	 * @throws Exception
	 */
	private void dealMessage(List<byte[]> messageList) throws Exception{
		//对消息内容进行处理
		try {
			RedisMessage message=(RedisMessage)BinaryTransferUtils.transferByteArrayToObject(messageList.get(1));
			logger.debug("该消息是内部对象,进行消息的预处理");
			String replyQueueName=message.getReplyQueueName();
			String uuid=message.getUuid();//需要回复的uuid
			Message mess=(Message)message.getObj();//真正的请求体
			String serviceName=message.getServiceName();//需要执行的服务名
			//使用一个虚假的寻求服务的方法
			if(Constants.W2N.equals(message.getDirection())) {
				//说明当前点为内网节点
				if ("DealService".equals(serviceName)) {
					String result=new DealService().invoke(mess.getName());
					if (message.isReply()) {//如果该消息需要进行回复
						Message replyme=new Message();
						replyme.setName(result);
						JedisUtils.sendObj(redisName,replyQueueName,null,replyme,uuid, Constants.N2W, false);
					}
				}
			}else if(Constants.N2W.equals(message.getDirection())) {
				//说明当前点为外网节点，接收的是回复的消息
				RedisReplyHandler.getInit().dealResponse(uuid, message);
			}
		} catch (Exception e) {
			try {
				logger.debug("该消息是测对象,内容为:"+new String(messageList.get(1))+"==");
			} catch (Exception e2) {
				logger.debug("异常信息:"+e);
			}
		}
	}
}
